package com.desay.pangoo.itmt.service.impl;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBody;

import com.desay.pangoo.common.lite.lang.web.BusinessException;
import com.desay.pangoo.itmt.controller.vo.rag.req.MessageReqVo;
import com.desay.pangoo.itmt.controller.vo.rag.req.UpdateChatTitleReqVo;
import com.desay.pangoo.itmt.controller.vo.rag.req.UpdateMessageContentReqVo;
import com.desay.pangoo.itmt.controller.vo.rag.resp.*;
import com.desay.pangoo.itmt.dto.*;
import com.desay.pangoo.itmt.entity.TechProjectInfoEntity;
import com.desay.pangoo.itmt.entity.rag.*;
import com.desay.pangoo.itmt.enums.AIRoleEnum;
import com.desay.pangoo.itmt.enums.AiVoteEnum;
import com.desay.pangoo.itmt.enums.ExceptionCodeEnum;
import com.desay.pangoo.itmt.repository.TechProjectInfoRepository;
import com.desay.pangoo.itmt.repository.rag.*;
import com.desay.pangoo.itmt.service.AIChatService;
import com.desay.pangoo.itmt.service.AsyncAiTaskService;
import com.desay.pangoo.itmt.service.LLMService;
import com.fasterxml.jackson.databind.ObjectMapper;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSON;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.log4j.Log4j2;

/**
 * AI 智能问答的核心业务逻辑
 *
 * @author uids1175
 */
@Log4j2
@Service
public class AIChatServiceImpl implements AIChatService {

	@Autowired
	private AsyncAiTaskService asyncAiTaskService;

	@Autowired
	private LLMService llmService;

	@Autowired
	private ChatEntityRepository chatRepo;

	@Autowired
	private MessageEntityRepository messageRepo;

	@Autowired
	private TechProjectInfoRepository techProjectInfoRepository;

	@Autowired
	private MessageRelationTPNEntityRepository messageRelationTpnEntityRepository;

	@Autowired
	private RestTemplate restTemplate;

	@Autowired
	private ObjectMapper objectMapper;

	@Autowired
	private HotProblemEntityRepository hotProblemEntityRepository;

	@Autowired
	private ProductDemandRepository productDemandRepository;

	/**
	 * 获取指定用户的历史聊天会话 1. 校验用户是否存在 2. 获取该用户下的聊天会话信息 3. 将聊天会话信息转换为VO对象并返回
	 *
	 * @param userId
	 *            用户ID
	 * @return ChatVO的列表
	 */
	@Override
	public List<ChatRespVo> listChats(String userId) {
		// 校验用户是否存在
		if (userId == null) {
			throw new BusinessException(ExceptionCodeEnum.RAG_USER_NOT_EXIST);
		}
		// 获取该用户的全部聊天历史
		Iterable<ChatEntity> allChats = chatRepo.findAllByUserId(userId);

		// 流化数据
		Stream<ChatEntity> chatsStream = StreamSupport.stream(allChats.spliterator(), false);

		// 将ChatEntity实体对象转换为VO对象
		return chatsStream.map(entity -> {
			ChatRespVo vo = new ChatRespVo();
			BeanUtils.copyProperties(entity, vo);
			return vo;
		}).toList();

	}

	/**
	 * 创建新的会话 1. 校验用户是否合法 2. 创建一个新的会话记录
	 *
	 * @param userId
	 *            用户ID
	 * @return 新建的ChatVO
	 */
	@Override
	public ChatRespVo createNewChat(String userId) {

		// 校验用户是否存在
		if (userId == null) {
			throw new BusinessException(ExceptionCodeEnum.RAG_USER_NOT_EXIST);
		}

		// 创建一个新的ChatEntity
		ChatEntity chatEntity = new ChatEntity();
		chatEntity.setChatTitle("New Chat");
		chatEntity.setUserId(userId);

		// 保存ChatEntity实体对象
		chatEntity = chatRepo.save(chatEntity);

		// 预热大模型接口
		LlmInitDTO initDTO = new LlmInitDTO();
		initDTO.setKeepAlive(-1);
		llmService.initLlm(initDTO);

		// 转换为VO对象
		return BeanUtil.toBean(chatEntity, ChatRespVo.class);
	}

	/**
	 * 获取某个会话下的聊天历史 1. 判断会话是否存在 2. 获取该会话下的历史消息
	 *
	 * @param chatId
	 *            用户存在的会话Id
	 * @return 该会话下的历史消息
	 */
	@Override
	public HistoryRespVo listMessagesByChatId(Long chatId) {

		// 判断chatId是否为NULL
		if (chatId == null) {
			throw new BusinessException(ExceptionCodeEnum.RAG_CHAT_NOT_EXIST);
		}

		// 判断是否存在对应的chatEntity
		ChatEntity chatEntity = chatRepo.findById(chatId).orElse(null);
		if (chatEntity == null) {
			throw new BusinessException(ExceptionCodeEnum.RAG_CHAT_NOT_EXIST);
		}

		// 根据chatId获取历史消息
		List<MessageEntity> allMessages = messageRepo.findAllByChatId(chatEntity.getChatId());

		// 没有历史消息 直接返回空对象
		if (allMessages.isEmpty()) {
			return new HistoryRespVo();
		}

		// 将历史消息转换为对应的MessageId列表
		List<Long> messageIdList = allMessages.stream().map(MessageEntity::getMessageId).toList();

		// 通过MessageId列表获取相关的ProductDemand信息
		List<ProductDemandEntity> productDemandEntityList = productDemandRepository.findAllByMessageIdIn(messageIdList);

		// 通过produceDemandId获取相关的RelationTpnEntity信息,并封装到对应的ProductDemandRespVo中
		List<ProductDemandRespVo> productDemandRespVoList = assembleRelationTpnEntity(productDemandEntityList);

		// 根据messageId将ProductDemand分组
		Map<Long, List<ProductDemandRespVo>> productDemandMap = productDemandRespVoList.stream()
				.collect(Collectors.groupingBy(ProductDemandRespVo::getMessageId));

		// 根据parentId将allMessages分组，以得到子消息IdList
		Map<Long, List<Long>> parentIdMap = allMessages.stream().collect(Collectors.groupingBy(
				MessageEntity::getParentId, Collectors.mapping(MessageEntity::getMessageId, Collectors.toList())));

		// 拼接成为history实体
		Map<Long, HistoryMessageItemVo> history = allMessages.stream().map(entity -> {
			HistoryMessageItemVo historyMessageItemVo = new HistoryMessageItemVo();
			BeanUtils.copyProperties(entity, historyMessageItemVo);
			historyMessageItemVo.setChildrenIds(
					parentIdMap.getOrDefault(historyMessageItemVo.getMessageId(), Collections.emptyList()));
			historyMessageItemVo.setProductDemandList(
					productDemandMap.getOrDefault(historyMessageItemVo.getMessageId(), Collections.emptyList()));
			return historyMessageItemVo;
		}).collect(Collectors.toMap(HistoryMessageItemVo::getMessageId, Function.identity()));

		// 用最新的message对话作为默认的message展示
		Long maxMessageId = history.keySet().stream().max(Long::compare).orElse(null);

		// 整个逻辑会保证一定存在messageId
		assert maxMessageId != null;
		LinkedList<HistoryMessageItemVo> itemList = new LinkedList<>();
		Long curMessageId = maxMessageId;

		// 组装最新的聊天对话
		while (curMessageId != 0) {
			HistoryMessageItemVo itemVo = history.get(curMessageId);
			if (itemVo == null) {
				break;
			}
			itemList.addFirst(itemVo);
			curMessageId = itemVo.getParentId();
		}

		// 构造Vo对象
		HistoryRespVo historyRespVo = new HistoryRespVo();
		historyRespVo.setMessages(itemList);
		historyRespVo.setHistory(history);
		return historyRespVo;

	}

	/**
	 * 为productDemandEntityList中的每一个元素装配相关的relationTpnEntity对象
	 *
	 * @param productDemandEntityList
	 *            产品需求列表
	 */
	private List<ProductDemandRespVo> assembleRelationTpnEntity(List<ProductDemandEntity> productDemandEntityList) {

		// 将ProductDemandEntity转换为对应的idList
		List<Long> productDemandIdList = productDemandEntityList.stream().map(ProductDemandEntity::getProductDemandId)
				.toList();

		// 将ProductDemandEntity转换为ProductDemandRespVo
		List<ProductDemandRespVo> resList = productDemandEntityList.stream()
				.map(entity -> BeanUtil.toBean(entity, ProductDemandRespVo.class)).toList();

		List<MessageRelationTpnEntity> relationTpnEntityList = messageRelationTpnEntityRepository
				.findAllByProductDemandIdIn(productDemandIdList);
		// 判断 relationTpnEntityList 是否为空
		if (relationTpnEntityList == null || relationTpnEntityList.isEmpty()) {
			return resList;
		}

		// 查询与MessageRelationTpnEntity有关的TechProjectInfo信息
		Set<String> ptnSet = relationTpnEntityList.stream().map(MessageRelationTpnEntity::getPtn)
				.collect(Collectors.toSet());

		Map<String, TechProjectInfoEntity> techProjectInfoMap = techProjectInfoRepository.findAllByTechPtnIn(ptnSet)
				.stream().collect(Collectors.toMap(TechProjectInfoEntity::getTechPtn, Function.identity()));

		// 将relationTpnEntityList按照productDemandId分组,并将relationTpnEntityList转换为对应的vo对象
		Map<Long, List<MessageRelationTpnRespVo>> relationTpnMap = relationTpnEntityList.stream().collect(
				Collectors.groupingBy(MessageRelationTpnEntity::getProductDemandId, Collectors.mapping(entity -> {
					// 将techProjectInfoEntity中的关键信息赋值给MessageRelationTpnRespVo
					TechProjectInfoEntity techProjectInfoEntity = techProjectInfoMap.get(entity.getPtn());
					MessageRelationTpnRespVo tpnVo = BeanUtil.toBean(entity, MessageRelationTpnRespVo.class);
					tpnVo.setPtnId(techProjectInfoEntity.getId());
					tpnVo.setTechLevel(techProjectInfoEntity.getTechLevel());
					tpnVo.setPriority(techProjectInfoEntity.getPriority());
					tpnVo.setTechMaturity(techProjectInfoEntity.getTechMaturity());
					tpnVo.setIsMajorProject(techProjectInfoEntity.getIsMajorProject());
					return tpnVo;
				}, Collectors.toList())));

		// 利用上面的分组数据，将resList中每个元素的tpnInfoList填充好数据
		for (ProductDemandRespVo vo : resList) {
			vo.setTpnInfoList(relationTpnMap.get(vo.getProductDemandId()));
		}

		return resList;

	}

	/**
	 * AI 问答内容 1. 处理Message 2. 交给算法端处理 3. 处理算法端返回的内容 4. 交给前端。
	 *
	 * @param historyMessages
	 *            历史消息列表
	 * @return AI生成的文本内容，流式数据，数据格式，一行一个JSON
	 */
	@Override
	public ResponseEntity<StreamingResponseBody> aiCompletion(List<MessageReqVo> historyMessages, Long chatId,
			String lang) {

		// 过滤掉总结消息
		historyMessages = historyMessages.stream().filter(vo -> vo.getIsSummary() == 0).toList();

		// 处理客户端消息，将其转换为AIMessageDTO并封装为AIMessageChatDTO
		List<LlmMessageDTO> messageDTOList = historyMessages.stream()
				.map(vo -> BeanUtil.toBean(vo, LlmMessageDTO.class)).toList();

		// 将消息数据封装为算法端支持的格式
		LlmCompletionDTO llmMessageDTO = new LlmCompletionDTO();
		llmMessageDTO.setMessages(messageDTOList);
		llmMessageDTO.setLang(lang);

		// 获取最新的一条消息的messageId
		MessageReqVo lastMessage = historyMessages.get(historyMessages.size() - 1);
		MessageEntity messageEntity = null;
		if (lastMessage.getMessageId() == null) {
			StreamMessageRespVo streamMessageRespVo = BeanUtil.toBean(lastMessage, StreamMessageRespVo.class);
			messageEntity = saveNewUserMessage(streamMessageRespVo);
		}

		// 获取Ai回答消息的parentId
		Long parentId = lastMessage.getMessageId();
		if (parentId == null && messageEntity != null) {
			parentId = messageEntity.getMessageId();
		}

		StreamingResponseBody streamingResponseBody = null;
		Long finalParentId = parentId;
		// streamingResponseBody = genStreamingBody(chatId, () ->
		// llmService.completion(llmMessageDTO), finalParentId, 0);
		StreamDataProcessor processor = new StreamDataProcessor();
		streamingResponseBody = processor.genStreamingBody(chatId, () -> llmService.completion(llmMessageDTO),
				finalParentId, 0);
		return ResponseEntity.ok().contentType(MediaType.APPLICATION_NDJSON).body(streamingResponseBody);
	}

	@Override
	public ResponseEntity<StreamingResponseBody> aiSummary(MessageReqVo message, Long chatId, String lang) {

		if (message == null || message.getIsSummary() == null) {
			throw new BusinessException(ExceptionCodeEnum.RAG_PARAMS_ERROR);
		}

		// 校验当前消息是AI消息
		if (!AIRoleEnum.ASSISTANT.getName().equals(message.getRole())) {
			throw new BusinessException(ExceptionCodeEnum.RAG_MESSAGE_NOT_AI);
		}
		// 获取这条AI消息的messageId作为总结消息的parentId
		Long parentId = message.getMessageId();
		// 然后生成流式数据
		LlmCompletionDTO llmCompletionDTO = new LlmCompletionDTO();
		LlmMessageDTO dto = BeanUtil.toBean(message, LlmMessageDTO.class);
		llmCompletionDTO.setMessages(Collections.singletonList(dto));
		llmCompletionDTO.setLang(lang);
		// StreamingResponseBody streamingResponseBody =
		// genStreamingBody(chatId,
		// () -> llmService.summary(llmCompletionDTO), parentId, 1);
		StreamDataProcessor processor = new StreamDataProcessor();
		StreamingResponseBody streamingResponseBody = processor.genStreamingBody(chatId,
				() -> llmService.summary(llmCompletionDTO), parentId, 1);
		return ResponseEntity.ok().contentType(MediaType.APPLICATION_NDJSON).body(streamingResponseBody);
	}

	private StreamingResponseBody genStreamingBody(Long chatId, Supplier<LlmCompletionRespDTO> aiBlock,
			Long finalParentId, Integer isSummary) {
		return outputStream -> {
			try {
				// 从AI生成的流式响应体中获取数据
				LlmCompletionRespDTO llmResponse = aiBlock.get();
				// 从请求头中获取 productDemand的相关数据
				List<ProductDemandRespVo> productDemandRespVoList = getProductDemandList(llmResponse.getMetaData());

				// 发送一条内容为空的AI message，并持久化
				StreamMessageRespVo message = createEmptyContentMessage(chatId, finalParentId, productDemandRespVoList,
						isSummary);
				writeln(outputStream, objectMapper.writeValueAsString(message));

				// 处理并转换AI生成的内容返回给客户端
				try (BufferedReader reader = new BufferedReader(new InputStreamReader(llmResponse.getInputStream()))) {
					String line;
					// 暂存 AI生成的文本内容
					List<PartContentMessageVo> partContentList = new ArrayList<>();
					while ((line = reader.readLine()) != null) {
						PartContentMessageVo partContentMessageVO = processAiPartResponse(line, message);
						writeln(outputStream, objectMapper.writeValueAsString(partContentMessageVO));
						partContentList.add(partContentMessageVO);
					}
					// 开启一个异步线程，持久化AI生成的Message
					asyncAiTaskService.updateNewAiMessageContent(message, partContentList);
				} catch (IOException e) {
					// 客户端中断连接的异常处理，比如将未处理完成的数据记录到日志
					log.error("Client connection interrupted", e);
					// 可以在这里添加数据的持久化逻辑，确保数据不丢失
				}
			} catch (Exception e) {
				throw new RuntimeException(e);
			}
		};
	}

	public class StreamDataProcessor {
		// 消息队列，用于暂存生成的数据
		private BlockingQueue<String> messageQueue = new LinkedBlockingQueue<>();

		// 定义异步数据生成任务
		private CompletableFuture<Void> dataGenerationTask;

		// 假设ExecutorService的线程数根据实际情况进行配置
		ExecutorService executorService = Executors.newFixedThreadPool(10);

		public StreamingResponseBody genStreamingBody(Long chatId, Supplier<LlmCompletionRespDTO> aiBlock,
				Long finalParentId, Integer isSummary) {
			// 使用线程安全的队列
			messageQueue = new LinkedBlockingQueue<>();
			dataGenerationTask = CompletableFuture
					.runAsync(() -> generateData(chatId, aiBlock, finalParentId, isSummary), executorService);

			return outputStream -> {
				try {
					// 判断 dataGenerationTask 线程执行结束
					while (!dataGenerationTask.isDone()) {
						// 优化队列操作，增加超时时间避免死锁
						String message = messageQueue.poll(1, TimeUnit.SECONDS);
						if (message != null) {
							writeln(outputStream, message);
						}
					}
					log.info("Close OutputStream ");
					// 在退出循环后关闭输出流
					outputStream.close();
				} catch (InterruptedException e) {
					Thread.currentThread().interrupt();
					log.info("Stream data processing interrupted");
				} catch (Exception e) {
					log.error("Failed to send streaming data", e);
				} finally {
					log.info("GenStreamingBody Stream data processing");
					if (dataGenerationTask != null) {
						log.info("Cancel Stream data processing");
						// 取消数据生成任务
						dataGenerationTask.cancel(true);
					}
					// 关闭ExecutorService
					executorService.shutdown();
				}
			};
		}

		private void generateData(Long chatId, Supplier<LlmCompletionRespDTO> aiBlock, Long finalParentId,
				Integer isSummary) {
			try {
				LlmCompletionRespDTO llmResponse = aiBlock.get();
				List<ProductDemandRespVo> productDemandRespVoList = getProductDemandList(llmResponse.getMetaData());
				StreamMessageRespVo message = createEmptyContentMessage(chatId, finalParentId, productDemandRespVoList,
						isSummary);
				messageQueue.put(objectMapper.writeValueAsString(message));

				try (BufferedReader reader = new BufferedReader(new InputStreamReader(llmResponse.getInputStream()))) {
					String line;
					List<PartContentMessageVo> partContentList = new ArrayList<>();
					while ((line = reader.readLine()) != null) {
						PartContentMessageVo partContentMessageVO = processAiPartResponse(line, message);
						messageQueue.put(objectMapper.writeValueAsString(partContentMessageVO));
						partContentList.add(partContentMessageVO);
					}
					log.info("Generate streaming body end");
					// 使用异步服务更新AI消息内容，考虑异常处理
					asyncAiTaskService.updateNewAiMessageContent(message, partContentList);
				} catch (IOException e) {
					log.error("Client connection interrupted", e);
				} catch (Exception e) {
					log.error("Unexpected error occurred", e);
				}
			} catch (Exception e) {
				log.error("Failed to generate streaming body", e);
			}
		}
	}

	/**
	 * 获取前10个热度最高的问题。 1. 从数据库中获取10个热度最高的问题。
	 *
	 * @return 热度问题列表
	 */
	@Override
	public List<HotProblemVo> getTop10HotProblem() {
		List<HotProblemEntity> list = hotProblemEntityRepository.findTop10ByOrderByHotDesc();
		// 如果没有热度问题，直接返回一个空集合
		if (list == null || list.isEmpty()) {
			return Collections.emptyList();
		}
		// 转换为Vo对象
		return list.stream().map(entity -> BeanUtil.toBean(entity, HotProblemVo.class)).toList();
	}

	/**
	 * 修改对应的vote状态，对应的备注
	 *
	 * @param code
	 *            1为点赞，-1点踩 0未操作/重置
	 * @param messageId
	 *            需要修改状态的消息Id
	 * @param comment
	 *            对应的消息备注
	 */
	@Override
	public void saveVoteStatusAndComment(Integer code, Long messageId, String comment) {

		// 获取MessageEntity
		MessageEntity messageEntity = messageRepo.findByMessageId(messageId);

		// 判断消息是否存在
		if (messageEntity == null) {
			throw new BusinessException(ExceptionCodeEnum.RAG_MESSAGE_NOT_EXIST);
		}

		// 校验MessageEntity 只允许Ai回答的消息可以修改
		if (!AIRoleEnum.ASSISTANT.getName().equals(messageEntity.getRole())) {
			throw new BusinessException(ExceptionCodeEnum.RAG_MESSAGE_NOT_AI);
		}

		// 根据code值修改对应的备注信息
		if (AiVoteEnum.LIKE.getCode().equals(code)) {
			messageEntity.setLikeComment(comment);
		}

		if (AiVoteEnum.DISLIKE.getCode().equals(code)) {
			messageEntity.setDislikeComment(comment);
		}

		messageEntity.setVote(code);

		// 保存MessageEntity
		messageRepo.save(messageEntity);

	}

	/**
	 * 根据聊天会话Id,清空会话中的历史消息
	 *
	 * @param chatId
	 *            聊天会话Id
	 */
	@Transactional(rollbackFor = Exception.class)
	@Override
	public void clearHistoryMessage(Long chatId) {
		// 判断会话是否存在
		ChatEntity chatEntity = chatRepo.findByChatId(chatId);
		if (chatEntity == null) {
			throw new BusinessException(ExceptionCodeEnum.RAG_CHAT_NOT_EXIST);
		}
		// 判断会话中是否存在历史消息
		long count = messageRepo.countByChatId(chatId);
		if (count == 0) {
			return;
		}
		// 如果存在会话消息就清空全部消息 需要确认是否是逻辑删除！！！
		messageRepo.deleteByChatId(chatId);

	}

	/**
	 * 根据MessageId更新Message的内容
	 *
	 * @param reqVo
	 *            参数实体
	 */
	@Override
	public void updateAiMessageContent(UpdateMessageContentReqVo reqVo) {
		// 校验参数: 消息ID不为空，内容不为空
		validUpdateMessageContentReqVo(reqVo);

		// 判断消息是否存在
		MessageEntity messageEntity = checkMessageExist(reqVo.getMessageId());

		// 判断消息是不是Ai回答的消息
		if (!messageEntity.getRole().equals(AIRoleEnum.ASSISTANT.getName())) {
			throw new BusinessException(ExceptionCodeEnum.RAG_MESSAGE_NOT_AI);
		}
		// 更新消息
		messageEntity.setContent(reqVo.getContent());
		messageRepo.save(messageEntity);
	}

	/**
	 * 根据ChatId修改聊天会话的标题
	 *
	 * @param reqVo
	 *            参数实体对象
	 */
	@Override
	public void updateChatTitle(UpdateChatTitleReqVo reqVo) {
		// 参数校验
		if (reqVo.getChatId() == null || StrUtil.isEmpty(reqVo.getTitle())) {
			throw new BusinessException(ExceptionCodeEnum.RAG_PARAMS_ERROR);
		}

		// 判断聊天会话是否存在
		ChatEntity chatEntity = chatRepo.findByChatId(reqVo.getChatId());
		if (chatEntity == null) {
			throw new BusinessException(ExceptionCodeEnum.RAG_CHAT_NOT_EXIST);
		}

		// 修改聊天会话
		chatEntity.setChatTitle(reqVo.getTitle());
		chatRepo.save(chatEntity);
	}

	/**
	 * 根据聊天会话Id删除一个聊天会话，以及这个聊天会话下的历史消息
	 *
	 * @param chatId
	 *            聊天会话Id
	 */
	@Transactional(rollbackFor = Exception.class)
	@Override
	public void deleteChatById(Long chatId) {
		// 校验参数
		if (chatId == null) {
			throw new BusinessException(ExceptionCodeEnum.RAG_PARAMS_ERROR);
		}
		// 判断会话是否存在
		ChatEntity chatEntity = chatRepo.findByChatId(chatId);
		if (chatEntity == null) {
			throw new BusinessException(ExceptionCodeEnum.RAG_CHAT_NOT_EXIST);
		}

		// 删除该会话以及对应的历史消息
		messageRepo.deleteByChatId(chatId);
		chatRepo.deleteByChatId(chatId);

	}

	/**
	 * 根据messageId删除某条消息，包括其子消息，逻辑删除
	 *
	 * @param messageId
	 *            某条消息Id
	 */
	@Transactional(rollbackFor = Exception.class)
	@Override
	public void deleteMessageById(Long messageId) {
		// 校验参数
		if (messageId == null) {
			throw new BusinessException(ExceptionCodeEnum.RAG_PARAMS_ERROR);
		}
		// 判断消息是否存在
		MessageEntity messageEntity = messageRepo.findByMessageId(messageId);
		if (messageEntity == null) {
			throw new BusinessException(ExceptionCodeEnum.RAG_MESSAGE_NOT_EXIST);
		}

		// 限制：只有最新的用户消息才能被删除，否则不允许删除。
		boolean res = isLastUserMessage(messageEntity);
		if (!res) {
			throw new BusinessException(ExceptionCodeEnum.RAG_MESSAGE_NOT_DELETE);
		}

		// 删除消息及其子消息
		messageRepo.deleteByMessageId(messageId);
		messageRepo.deleteByParentId(messageEntity.getMessageId());
	}

	/**
	 * 判断是否是最新的用户消息
	 *
	 * @param messageEntity
	 *            一个消息实体
	 * @return true 标识是最新的，false标识不是
	 */
	private boolean isLastUserMessage(MessageEntity messageEntity) {
		if (messageEntity == null) {
			return false;
		}

		if (!messageEntity.getRole().equals(AIRoleEnum.USER.getName())) {
			return false;
		}

		// 查询比当前messageId大的role=user以及chatId等于当前消息的chatId的消息集合
		List<MessageEntity> list = messageRepo.findAllByChatIdAndRoleAndMessageIdGreaterThan(messageEntity.getChatId(),
				messageEntity.getRole(), messageEntity.getMessageId());

		if (list == null || list.isEmpty()) {
			return true;
		}
		// 判断消息集合中的parentId是否与原消息相同，如果都相同则说明是同级且最新的，那么可以删除
		Set<Long> set = list.stream().map(MessageEntity::getParentId).collect(Collectors.toSet());
		return set.size() <= 1;
	}

	/**
	 * 判断消息实体是否存在
	 *
	 * @param messageId
	 *            消息Id
	 * @return 数据库中存在的消息实体
	 */
	private MessageEntity checkMessageExist(Long messageId) {
		MessageEntity entity = messageRepo.findByMessageId(messageId);
		if (entity == null) {
			throw new BusinessException(ExceptionCodeEnum.RAG_MESSAGE_NOT_EXIST);
		}
		return entity;
	}

	/**
	 * 校验参数的合法性: 消息ID不为空，内容不为空
	 *
	 * @param reqVo
	 *            参数实体
	 */
	private void validUpdateMessageContentReqVo(UpdateMessageContentReqVo reqVo) {
		if (reqVo == null || reqVo.getMessageId() == null) {
			throw new BusinessException(ExceptionCodeEnum.RAG_PARAMS_ERROR);
		}
		if (StrUtil.isEmpty(reqVo.getContent())) {
			throw new BusinessException(ExceptionCodeEnum.RAG_PARAMS_ERROR);
		}
	}

	/*
	 * ================================== private区
	 * ==================================
	 */

	/**
	 * 处理AI返回的消息内容，封装成统一的格式
	 *
	 * @param line
	 *            AI生成的一行JSON数据
	 * @param message
	 *            新生成的AI Message
	 * @return 包含部分文本内容的对象
	 */
	private static PartContentMessageVo processAiPartResponse(String line, StreamMessageRespVo message) {
		// 处理消息
		JSON jsonObj = JSONUtil.parse(line);
		String content = (String) jsonObj.getByPath("response");
		if (content == null) {
			content = (String) jsonObj.getByPath("message.content");
		}
		Boolean done = (Boolean) jsonObj.getByPath("done");
		if (done == null) {
			content = (String) jsonObj.getByPath("message.done");
		}

		// 转换为PartContentMessageVO的格式
		PartContentMessageVo partContentMessageVO = new PartContentMessageVo();
		partContentMessageVO.setMessageId(message.getMessageId());
		partContentMessageVO.setContent(content);
		partContentMessageVO.setDone(done);
		partContentMessageVO.setIsPart(true);
		return partContentMessageVO;
	}

	/**
	 * 从请求头中，获取相关Tech Project Info数据 metadata的格式： { "is_in_lib": 1,
	 * "project_metadata": { "产品需求1": [["PTNCNB21036", 0.6835251187227644],
	 * ["PTNPNA21047", 0.39787992048216886]], "产品需求2": [["PTNCNB21036",
	 * 0.6835251187227644], ["PTNPNA21047", 0.39787992048216886]] } }
	 *
	 * @param metadata
	 *            算法端返回的元数据
	 * @return 与AI生成的消息有关的 Tech Project Info数据列表
	 */
	private List<ProductDemandRespVo> getProductDemandList(String metadata) {

		if (metadata == null) {
			return Collections.emptyList();
		}
		return parseProjectMeta(metadata);
	}

	/**
	 * 从metadata中解析出真正元数据信息
	 * 
	 * @param metadata
	 *            一个非空字符串，不一定保证是一个JSON格式的字符串
	 * @return 解析出来的真正的元数据JSON对象
	 */
	private static final String PROJECT_METADATA_FIELD = "project_metadata";

	private List<ProductDemandRespVo> parseProjectMeta(String metadata) {
		Map<String, Object> metadataJsonMap = null;
		try {
			JSONObject jsonObject = JSONUtil.parseObj(metadata);
			metadataJsonMap = jsonObject.getByPath(PROJECT_METADATA_FIELD, Map.class);
			if (metadataJsonMap == null) {
				throw new BusinessException(ExceptionCodeEnum.RAG_METADATA_PARSE_ERROR);
			}
		} catch (Exception e) {
			// JSON 解析失败，包含原始异常信息
			throw new BusinessException(ExceptionCodeEnum.RAG_METADATA_PARSE_ERROR);
		}

		// 遍历metadataMap
		List<ProductDemandRespVo> listVo = new ArrayList<>();
		for (String productDemandName : metadataJsonMap.keySet()) {
			Object jsonObj = metadataJsonMap.get(productDemandName);
			JSONObject jsonData = (JSONObject) jsonObj;

			// 解析JsonArray封装LlmMetadataItemDTO对象
			JSONArray projectList = (JSONArray) jsonData.get("project_list");
			String description = jsonData.getStr("description");
			List<LlmMetadataItemDTO> items = new ArrayList<>();
			for (Object metadataArrObj : projectList) {
				if (!(metadataArrObj instanceof JSONArray)) {
					continue;
				}
				JSONArray metadataArr = (JSONArray) metadataArrObj;
				if (metadataArr.size() != 2) {
					continue;
				}
				List<LlmMetadataItemDTO> parsedItems = parseItems(metadataArr);
				items.addAll(parsedItems);
			}
			ProductDemandRespVo vo = new ProductDemandRespVo();
			vo.setDescription(description);
			vo.setProductDemandName(productDemandName);
			Map<String, TechProjectInfoEntity> techInfoMap = getTechInfoMap(items);
			List<MessageRelationTpnRespVo> tpnInfoList = convertToTpnInfoList(items, techInfoMap);
			vo.setTpnInfoList(tpnInfoList);
			listVo.add(vo);
		}

		return listVo;
	}

	private Map<String, TechProjectInfoEntity> getTechInfoMap(List<LlmMetadataItemDTO> list) {
		Set<String> ptnSet = list.stream().map(LlmMetadataItemDTO::getTechPtn).collect(Collectors.toSet());
		List<TechProjectInfoEntity> techInfoList = techProjectInfoRepository.findAllByTechPtnIn(ptnSet);
		return techInfoList.stream().collect(Collectors.toMap(TechProjectInfoEntity::getTechPtn, Function.identity()));
	}

	private List<MessageRelationTpnRespVo> convertToTpnInfoList(List<LlmMetadataItemDTO> itemList,
			Map<String, TechProjectInfoEntity> techInfoMap) {
		if (itemList == null || itemList.isEmpty()) {
			return Collections.emptyList();
		}
		return itemList.stream().distinct().filter(item -> techInfoMap.containsKey(item.getTechPtn())).map(item -> {
			TechProjectInfoEntity techEntity = techInfoMap.get(item.getTechPtn());
			MessageRelationTpnRespVo messageRelationTpnRespVo = BeanUtil.toBean(techEntity,
					MessageRelationTpnRespVo.class);
			messageRelationTpnRespVo.setPtn(techEntity.getTechPtn());
			messageRelationTpnRespVo.setScore(item.getScore());
			messageRelationTpnRespVo.setPtnId(techEntity.getId());
			return messageRelationTpnRespVo;
		}).collect(Collectors.toList());
	}

	/**
	 * [ "PTNOPS23003", 0.8514594320170558 ], [
	 * "PTNSNA23009.01;PTNSNA23009.02;PTNSNA23009.03", 0.48000639421359054 ],
	 * 
	 * @param metadataArr
	 *            一定保证metadataArr是一个Json数组，但是Json数组存在两种情况
	 * @return 一个LlmMetadataItemDTO列表
	 */
	private static final String DIVIDER_IDENTIFIER = ";";

	private List<LlmMetadataItemDTO> parseItems(JSONArray metadataArr) {
		String ptns = metadataArr.getStr(0);
		String[] ptnArr = ptns.split(DIVIDER_IDENTIFIER);
		return Stream.of(ptnArr).map(ptn -> new LlmMetadataItemDTO(ptn, metadataArr.getBigDecimal(1).doubleValue()))
				.toList();
	}

	/**
	 * 创建一个role=assistant的空消息
	 *
	 * @return 内容为空的AIMessage
	 */
	private StreamMessageRespVo createEmptyContentMessage(Long chatId, Long parentId, List<ProductDemandRespVo> list,
			Integer isSummary) {
		// 设置除content的全部属性,并持久化
		MessageEntity entity = new MessageEntity();
		entity.setChatId(chatId);
		entity.setContent("");
		entity.setRole(AIRoleEnum.ASSISTANT.getName());
		entity.setVote(0);
		entity.setParentId(parentId);
		entity.setIsSummary(isSummary);
		messageRepo.save(entity);

		// 转换为Vo对象
		StreamMessageRespVo vo = BeanUtil.toBean(entity, StreamMessageRespVo.class);
		vo.setDone(false);
		vo.setProductDemandList(list);
		return vo;
	}

	/**
	 * 写入一行数据
	 *
	 * @param outputStream
	 *            一个输出流
	 * @param content
	 *            一个文本数据
	 */
	private void writeln(OutputStream outputStream, String content) {
		try {
			outputStream.write(content.getBytes(StandardCharsets.UTF_8));
			outputStream.write("\n".getBytes(StandardCharsets.UTF_8));
			outputStream.flush();
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
	}

	private MessageEntity saveNewUserMessage(StreamMessageRespVo message) {
		// 判断是不是用户发送的消息
		if (!AIRoleEnum.USER.getName().equals(message.getRole())) {
			return null;
		}

		// 保存消息
		MessageEntity entity = BeanUtil.toBean(message, MessageEntity.class);
		return messageRepo.save(entity);
	}

}
